package mq;

import com.frxs.framework.integration.rocketmq.listener.AbstractMessageListenerConcurrently;
import com.zwl.yyyy.common.util.CheckUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.List;

@Configuration
@Slf4j
public class MqConsumerTagConfig {

	@Autowired(required = false)
	private List<DefaultMQPushConsumer> defaultMQPushConsumerList;

	@Autowired
	private MqConsumerWhitelistHandler mqConsumerWhitelistHandler;

	@PostConstruct
	public void groupAddTag() {
		if (CheckUtil.isEmpty(defaultMQPushConsumerList)) {
			return;
		}
		EmptyConsumerListener emptyConsumerListener = new EmptyConsumerListener();
		defaultMQPushConsumerList.forEach(e -> {
			String consumerGroup = e.getConsumerGroup();

			String mqConsumerGroupTag = mqConsumerWhitelistHandler.getMqConsumerGroupTag();
			consumerGroup = consumerGroup + mqConsumerGroupTag;
			e.setConsumerGroup(consumerGroup);

			MessageListener messageListener = e.getMessageListener();
			if (null != messageListener) {
				String listenerClassName = messageListener.getClass().getName();
				boolean disable = mqConsumerWhitelistHandler.isDisable(listenerClassName);
				if (disable) {
					log.info("禁掉的mq监听: listenerClassName:{}", listenerClassName);
					//不能形成阻塞消息，要让在本地消费掉，但是不走业务逻辑
					e.setMessageListener(emptyConsumerListener);
				}
			}
		});
	}

	private class EmptyConsumerListener extends AbstractMessageListenerConcurrently {

		/**
		 *
		 * @param list
		 * @param consumeConcurrentlyContext
		 * @return
		 */
		@Override
		public ConsumeConcurrentlyStatus execute(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
			log.debug("EmptyConsumerListener.list = " + list);
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}

}
